package com.wenx.kafka;

import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 * kafka生产数据
 */
public class CollectKafkaProducer {

    public static void main(String[] args) throws InterruptedException {
        String topic = "topic1";
        CollectKafkaProducer collectKafkaProducer = new CollectKafkaProducer(topic);
        for (int i = 0; i < 10; i++) {
            Map<String,String> map = new HashMap<>();
            map.put("id",i+"");
            map.put("name","张三"+i);
            collectKafkaProducer.send(map,false);
        }
        //主线程等待
        TimeUnit.DAYS.sleep(1L);
    }

    //生产者
    private final KafkaProducer<String, String> producer;

    //主题
    private final String topic;

    public CollectKafkaProducer(String topic) {
        Properties props = new Properties();
        props.put("bootstrap.servers","192.168.70.235:9092");
        props.put("client.id","demo-producer-test");
        // kafka 序列化配置：
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        this.producer = new KafkaProducer<>(props);
        this.topic = topic;
    }

    /**
     * 发送消息
     * @param message
     * @param syncSend
     */
    public void send(Object message, Boolean syncSend) {

        if(syncSend) {
            //同步发送
            producer.send(new ProducerRecord<>(topic, JSON.toJSONString(message)));
        } else {
            //异步发送
            producer.send(new ProducerRecord<>(topic, JSON.toJSONString(message)), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (e != null) {
                        System.err.println("Unable to write to Kafka in CollectKafkaProducer [" + topic + "] exception: " + e);
                    }
                }
            });
        }

    }

    /**
     * 关闭
     */
    public void close() {
        producer.close();
    }


}
